package com.headfirst.dmp.report.AreanalyzeRpt

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 分析统计按照不同地域的各项指标
  *
  * 使用工具：saprk-sql
  */
object AreaAnalyzeRPT {


  def main(args: Array[String]): Unit = {

    //1.验证参数
    if (args.length != 2) {
      print(
        """
          |com.headfirst.dmp.report.AreaAnalyzeRPT
          |需要参数：
          |       logInputPath
        """.stripMargin)
      sys.exit(-1)
    }

    //2.接受参数
    val Array(logInputPath) = args

    //3.创建sparksession对象
    val conf = new SparkConf()
    conf.setMaster("local[*]") //本地测试使用local，提交到集群则注释掉该配置
    conf.setAppName("AreaAnalyzeRPT")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）

    val session: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    //4.读取文件
    val logFile: DataFrame = session.read.parquet(logInputPath)

    //5.注册表
    logFile.createTempView("v_log")

    //6.执行sql语句（此处使用case when ，这个是标准的sql语法）\
    // 原始请求数，有效请求数..这些是起的别名
    val result: DataFrame = session.sql(
      """
        |select
        |sum (case when requestmode=1 and processnode>=1 then 1 else 0 end)  原始请求数
        |sum (case when requestmode=1 and processnode>=2 then 1 else 0 end)  有效请求数
        |sum (case when requestmode=1 and processnode =3 then 1 else 0 end)  广告请求数
        |sum (case when adplatformproviderid>=100000 and iseffective=1 and isBilling =1 and isBid=1 and ADOrderid!=0 then 1 else 0 end) 参与竞价的次数
        |sum (case when adplatformproviderid>=100000 and iseffective=1 and isBilling =1 and isWin=1 then 1 else 0 end) 参与竞价的次数
        |sum (case when requestmode=2 and iseffective=1) 展示数（广告主）
        |sum (case when requestmode=3 and iseffective=1) 点击数（广告主）
        |case when adplatformproviderid>=100000 and iseffective=1 and isBilling=1 and isWin=1 and ADOrderid>=200000 and adcreativeid>=>=200000  then WinPrice/1000 else 0 end  DSP广告消费
        |case when adplatformproviderid>=100000 and iseffective=1 and isBilling=1 and isWin=1 and ADOrderid>=200000 and adcreativeid>=>=200000  then adPayment/1000 else 0 end  DSP广告成本
        |from v_log
        |group by provincename,cityname
      """.stripMargin)


    //7.将数据写入Mysql数据(此处利用读取配置文件信息)
    val config: Config = ConfigFactory.load()
    val prop = new Properties()
    prop.setProperty("user", config.getString("jdbc.user"))
    prop.setProperty("password", config.getString("jdbc.password"))

    result.write.mode("append").jdbc(config.getString("jdbc.url"), config.getString("jdbc.tableName"), prop)

    //8.关流
    session.stop()

  }


}
